package com.mark.article.stream;

import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.mark.common.kafka.KafkaTopicConfig;
import com.mark.common.kafka.messages.UpdateArticleMessage;
import com.mark.common.kafka.messages.app.ArticleVisitStreamMessage;
import com.mark.common.kafkastream.KafkaStreamListener;
import com.mark.model.mess.app.ArticleVisitStreamDto;
import lombok.extern.log4j.Log4j2;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.kstream.*;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

/**
 * @Description :  热门文章流处理程序
 * @Author :  Markburt
 * @CreateDate :  2020/2/28$ 下午 07:45$
 * @UpdateUser :  Markburt
 * @UpdateDate :  2020/2/28$ 下午 07:45$
 * @UpdateRemark :  Project Build
 * @Version :  1.0
 */
@Component
@Log4j2
public class HotArticleStreamHandler implements KafkaStreamListener<KStream<?, String>> {


    @Autowired
    private KafkaTopicConfig kafkaTopicConfig;

    @Autowired
    private ObjectMapper mapper;


    /**
     * 监听的类型
     *
     * @return
     */
    @Override
    public String listenerTopic() {
        return kafkaTopicConfig.getArticleUpdateBus();
    }

    /**
     * 处理结果发送的类
     *
     * @return
     */
    @Override
    public String sendTopic() {
        return kafkaTopicConfig.getArticleIncrHandle();
    }

    /**
     * 对象处理逻辑
     *
     * @param stream
     * @return
     */
    @Override
    public KStream<?, String> getService(KStream<?, String> stream) {
        return stream.map(
                (key, val) -> {
                    UpdateArticleMessage value = format(val);
                    System.out.println(value);
                    //likes:1
                    return new KeyValue<>(
                            value.getData().getArticleId().toString(),
                            value.getData().getType().name() + ":" +
                                    value.getData().getAdd()
                    );
                }
        ).groupByKey().windowedBy(TimeWindows.of(10000)).aggregate(
                new Initializer<String>() {
                    @Override
                    public String apply() {
                        return "COLLECTION:0,COMMENT:0,LIKES:0,VIEWS:0";
                    }
                },
                new Aggregator<Object, String, String>() {
                    @Override
                    public String apply(Object aggKey, String value, String aggValue) {
                        //类似于  likes:1
                        value = value.replace(
                                "UpdateArticle(", "").replace(")",
                                ""
                        );
                        String valAry[] = value.split(":");
                        if ("null".equals(valAry[1])) {
                            return aggValue;
                        }
                        //"COLLECTION:0,COMMENT:0,LIKES:0,VIEWS:0";
                        //String[] aggArr = aggValue.split(",");

                        int col = 0, com = 0, lik = 0, vie = 0;
                        if ("LIKES".equalsIgnoreCase(valAry[0])) {
                            lik += Integer.valueOf(valAry[1]);
                        }
                        if ("COLLECTION".equalsIgnoreCase(valAry[0])) {
                            col += Integer.valueOf(valAry[1]);
                        }
                        if ("COMMENT".equalsIgnoreCase(valAry[0])) {
                            com += Integer.valueOf(valAry[1]);
                        }
                        if ("VIEWS".equalsIgnoreCase(valAry[0])) {
                            vie += Integer.valueOf(valAry[1]);
                        }

                        return String.format("COLLECTION:%d,COMMENT:%d,LIKES:%d,VIEWS:%d", col, com, lik, vie);
                    }
                }, Materialized.as("count-article-num-miukoo-1")
        ).toStream().map((key, value) -> {
            return new KeyValue<>(
                    key.key().toString(),
                    formatObj(key.key().toString(),
                            value)
            );
        });
    }


    /**
     * 格式转换
     * @param val
     * @return
     */
    private UpdateArticleMessage format(String val) {
        UpdateArticleMessage updateArticleMessage = null;
        try {
            updateArticleMessage = mapper.readValue(val, UpdateArticleMessage.class);
        } catch (IOException e) {
            e.printStackTrace();
        }
        return updateArticleMessage;
    }

    /**
     * Object 格式处理
     * @param articleId
     * @param value
     * @return
     */
    private String formatObj(String articleId, String value) {
        String ret = "";
        ArticleVisitStreamMessage temp = new ArticleVisitStreamMessage();
        ArticleVisitStreamDto dto = new ArticleVisitStreamDto();
        String regEx = "COLLECTION:(\\d+),COMMENT:(\\d+),LIKES:(\\d+),VIEWS:(\\d+)";
        Pattern pat = Pattern.compile(regEx);
        Matcher mat = pat.matcher(value);
        if (mat.find()) {
            dto.setCollect(Long.valueOf(mat.group(1)));
            dto.setCommont(Long.valueOf(mat.group(2)));
            dto.setLike(Long.valueOf(mat.group(3)));
            dto.setView(Long.valueOf(mat.group(4)));
        } else {
            dto.setCollect(0);
            dto.setCommont(0);
            dto.setLike(0);
            dto.setView(0);
        }
        dto.setArticleId(Integer.valueOf(articleId));
        temp.setData(dto);
        try {
            ret = mapper.writeValueAsString(temp);
        } catch (JsonProcessingException e) {
            e.printStackTrace();
        }
        return ret;
    }
}
